package io.wen.bd.s7m1

import java.sql.Connection

import com.alibaba.druid.pool.{DruidDataSource, DruidPooledConnection}
import org.apache.spark.sql.ForeachWriter

class MysqlWriter extends ForeachWriter[BusInfo] {
  var conn: Connection = _

  override def open(partitionId: Long, epochId: Long): Boolean = {
    conn = MysqlWriter.getConnection
    true
  }

  override def process(busInfo: BusInfo): Unit = {
    val lglat = busInfo.lglat

    val id = busInfo.deployNum + busInfo.plateNum + busInfo.timeStr
    val arr: Array[String] = lglat.split("_")

    val ps = conn.prepareStatement("insert into car_info (id, lng, lat) values (?,?,?)")
    ps.setObject(1, id)
    ps.setObject(2, arr(0))
    ps.setObject(3, arr(1))
    val result = ps.executeUpdate()
    println("MySql Writer process reuslt: " + result)
    ps.close()
  }

  override def close(errorOrNull: Throwable): Unit = {
    conn.close()
  }
}


object MysqlWriter {
  val datasoure: DruidDataSource = new DruidDataSource()
  datasoure.setUrl("jdbc:mysql://hadoop03:3306/lg_logstic")
  datasoure.setDriverClassName("com.mysql.jdbc.Driver")
  datasoure.setUsername("root")
  datasoure.setPassword("Ven3.06+")

  def getConnection: DruidPooledConnection = {
    datasoure.getConnection()
  }
}
